package org.diy.traffic.sources;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.TimeZone;
import java.util.logging.Level;

import org.apache.commons.httpclient.HttpStatus;
import org.diy.traffic.daemon.AirTrafficController;
import org.diy.traffic.sinks.DiyFluSink;
import org.diy.traffic.updates.TrafficUpdate;
import org.diy.traffic.updates.TrafficUpdate.UnknownDataException;
import org.diy.traffic.util.TLog;

import winterwell.jtwitter.TwitterException;
import winterwell.jtwitter.Twitter.Status;

public class DiyFluSource extends FeedSource {

	private static String sourceType = "DIYFLU_SOURCE";

	private static final int DEFAULT_GO_DAYS_BACK = 45;
	
	private static String mysqlDbName;
	private static String mysqlUsername;
	private static String mysqlPassword;
	private static int goDaysBack;
	
	private boolean UPDATE_IN_PROGRESS = false;
	
	// this should really be parameterized!
	private static int SOURCE_UPDATE_INTERVAL  = 600; // should be ten minites == 600s
	
	// this is the default; it is updated via the udpateInterval config parameter
	private static int FNCTL_PER_CITY_UPDATE_S = 3600; // each city updates every hour
	
	private Connection dbConn = null;
	
	private int lastCheckedCityId = -1;
	
	public DiyFluSource() throws TrafficSourceException {
		//this(TrafficSource.DEFAULT_UPDATE_FREQUENCY_S);
		this(SOURCE_UPDATE_INTERVAL); // every ten minutes is when inputdaemon tells me to update....
	}

	public DiyFluSource(int updateInterval) throws TrafficSourceException {
		super(sourceType, "http://search.twitter.com/search.json?", updateInterval);
		//this.sourceName = sourceType + "_" + username;
		initDB();
		
		// do this so that last check isn't 0, so that it won't think that all the cities are new.... 
		//this.setLastCheckToNow();
		this.setLastCheck( System.currentTimeMillis() - (1000*(updateInterval+2)) );
	}
	
	public void initDB() throws TrafficSourceException {
		try {
			dbConn = DriverManager.getConnection(
					String.format("jdbc:mysql://localhost/%s?user=%s&password=%s", mysqlDbName, mysqlUsername, mysqlPassword)
					);
		} catch (SQLException e) {
			TLog.log(Level.SEVERE, e);
			throw new TrafficSourceException(String.format("Can't connect to the database with dbname: '%s', username: '%s', password: '%s'",
					mysqlDbName, mysqlUsername, mysqlPassword));

		}
	}

	public static DiyFluSource createFromConfigData(HashMap<String, String> configParams) throws TrafficSourceException {
		mysqlDbName   = configParams.get("db");
		mysqlUsername = configParams.get("username");
		mysqlPassword = configParams.get("password");
		try {
			goDaysBack    = Integer.parseInt(configParams.get("history"));
		} catch (Exception e) {
			goDaysBack = DEFAULT_GO_DAYS_BACK;
		}
			
		
	
		if ((mysqlDbName == null) || (mysqlUsername == null) || (mysqlPassword == null)) {
			TLog.log(Level.SEVERE, "DiyFluSource requires a 'db', 'username', and 'password' parameter for configuration! None provided.");
			return null;
		}
		
		String updateInterval = configParams.get("updateInterval");
		if (updateInterval != null) {
			FNCTL_PER_CITY_UPDATE_S = Integer.parseInt(updateInterval);
		}
		
		return new DiyFluSource();
	}
	
	
	@Override
	public List<TrafficUpdate> doCheck(boolean updateLastCheck)
			throws TrafficSourceException {

		/** add this back in when threads are working...
		this.UPDATE_IN_PROGRESS = true;
		**/
		
		/**
		 * Description of Algorithm
		 * per_city_update_interval = the interval on which any given city should be updated
		 * source_update_interval   = the interval on which diyflusource updates
		 * 
		 * i.e. there are 20 cities (num_cities=20)[GIVEN]
		 * you want each one to update every 60 minutes (per_city_update=60m) [VARIABLE]
		 * you want the source_update_interval to run every 10 minutes (source_update=10m)[VARIABLE]
		 * 
		 * per_city_update / source_update == number of times it will run in the per_city_update interval (num_runs)
		 * num_cities / num_runs == the number of cities per run to achieve the per_city_update (recalc each round based on num_cities)
		 */
		
		
		
		List<TrafficUpdate> update_list = new ArrayList<TrafficUpdate>();
		
		Date last_check_date = new Date(this.getLastCheck());
		String last_check_str = getDateInMySQLFormat(last_check_date);
		
		ResultSet next_n_rs = null;
		Statement next_n_stmt = null;
		
		ResultSet since_rs = null;
		Statement since_stmt = null;
		
		Statement update_stmt = null;
		
		DiyFluSink.refreshBlackListTerms();
		
		try {
			
			String SPECIAL_WHERE_CLAUSE = " (country=\"Canada\" OR country=\"USA\" OR " +
					"country=\"Australia\" OR country=\"United Kingdom\" OR country=\"Ireland\" " +
					"OR country=\"Tawain\" OR country=\"India\") ";
			
			Statement countStmt = dbConn.createStatement();
			ResultSet countRs   = countStmt.executeQuery("SELECT COUNT(*) FROM cities");
			countRs.first();
			int cityCount = countRs.getInt(1);
			
			countRs.close();
			countStmt.close();
			
			
			
			
			//System.err.println("number of citeis is " + cityCount);
			
			// interval specified in config file on which each city should be updated
			int CITY_UPDATE_S = (FNCTL_PER_CITY_UPDATE_S); 
			//System.err.println("my intended update frequency is " + CITY_UPDATE_S + " seconds.");
			
			// this TrafficSource updates every 600s = 10minutes
			int SOURCE_UPDATE_S      = SOURCE_UPDATE_INTERVAL; 
			//System.err.println("this source updates every " + SOURCE_UPDATE_S + " seconds");
			
			// for example, if you want each city to update every hour, and this source will be checked
			// every ten minutes, then this source will be checked 6 times within that interval...
			double NUM_CHECKS = CITY_UPDATE_S / SOURCE_UPDATE_S;
			//System.err.println("This source will be checked " + NUM_CHECKS + " time within the intended ferquency");
			
			// ake this a MEMBER FIELD and set it by default to 5
			// ... therefore, you must check 1/6th of the cities with each run!
			int NUM_CITIES_PER_CHECK = (int) Math.ceil(cityCount / NUM_CHECKS);
			TLog.log(Level.FINE, 
					String.format("Will check %d of %d cities this update", NUM_CITIES_PER_CHECK , cityCount)
					);
			
			
			next_n_stmt = dbConn.createStatement();
	    	
			next_n_rs = next_n_stmt.executeQuery(
					String.format("SELECT * FROM cities WHERE id>=%d ORDER BY id LIMIT %d",
							lastCheckedCityId+1, NUM_CITIES_PER_CHECK )
					);
			
			// if there are no results in the result set!
			if (!next_n_rs.last()) {
				lastCheckedCityId=-1;
				next_n_rs = next_n_stmt.executeQuery(
						String.format("SELECT * FROM cities WHERE id>%d ORDER BY id LIMIT %d",
								lastCheckedCityId, NUM_CITIES_PER_CHECK )
						);
			} else {
				next_n_rs.beforeFirst();
			}
			

			// need to put this here before all the work is done (which could take a while), to avoid creating a race condition....
			if (updateLastCheck) 
				this.setLastCheckToNow();
			
			boolean has_next = next_n_rs.next();
			
			TweetComm tweet_comm = new TweetComm(this);
			Phrase[] phrase_array = fetchAllPhrases();
			
			// iterating through all the cities being checked for this check!
			while (has_next){
				String city_name = next_n_rs.getString("name");
				String country_name = next_n_rs.getString("country");
				String created_at  = next_n_rs.getString("created_at");
				String updated_at = next_n_rs.getString("updated_at");
				long city_id = next_n_rs.getLong("id");
				
				String[] lat_long;
				try {
					lat_long = this.fetchLatLongForCity(city_name);
				} catch (TrafficSourceException tse) {
					TLog.log(Level.WARNING, tse);
					has_next = next_n_rs.next();
					continue;
				}
				
				
				SimpleDateFormat date_formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.getDefault());
				Date created_at_date;
				try {
					created_at_date = date_formatter.parse(created_at);
				} catch (ParseException e) {
					TLog.log(Level.WARNING, e);
					has_next = next_n_rs.next();
					continue;
				}
				
				City a_city = new City(city_name, country_name, city_id, lat_long[0], lat_long[1]); 
				
				
				//TODO for each city, see what languages it uses, then go through this whole thing once for
				// each language. fetchAllPhrasesForCity(id)
				// get all the phrases, group by language, break this up into sets and issue one query for each with the 'lang' param
				// maybe each Phrase needs a 'lang' field?
				
				
				// if the last time you checked was before this city was added ==
				// the city is new
				//if (last_check_date.compareTo(created_at_date) <= 0) {
				if (created_at.compareTo(updated_at) == 0) {
					TLog.log(Level.INFO, String.format("****Found new city: '%s', getting %s days of historical data.", city_name, goDaysBack));
					
					/**
					Thread t = new Thread(gg);
					t.start();
					**/
					
					//actually doesn't add anything!
					update_list.addAll(tweet_comm.getPhrasesSinceDaysAgo(phrase_array, a_city, goDaysBack, true));
					
				} else {
					TLog.log(Level.FINE, String.format("Updating recent mention of all terms for city '%s' (%d)", city_name, city_id));
					
					since_stmt = dbConn.createStatement();
					//TODO This is a really inefficient way of doing this!!
					since_rs   = since_stmt.executeQuery(
							String.format("select MAX(source_id) FROM mentions WHERE city_id=%d", (int)city_id) 
							);
					
					since_rs.first();
					String res = since_rs.getString("MAX(source_id)");
					//if (res.compareTo("NULL")==0) {
					if (res == null) {
						TLog.log(Level.FINER, "(Fetching all of today's sicktweets)");
						//update_list.addAll(tweet_comm.getPhrasesSinceToday(phrase_array, a_city));
						tweet_comm.addUpdatesToQueue(tweet_comm.getPhrasesSinceToday(phrase_array, a_city));
				
					} else {
						int since_id = Integer.parseInt(res);
						TLog.log(Level.FINER, String.format("(Fetching sicktweets only since id %d)", since_id));
						//update_list.addAll(tweet_comm.getPhrasesSinceId(phrase_array, a_city, since_id));
						tweet_comm.addUpdatesToQueue(tweet_comm.getPhrasesSinceId(phrase_array, a_city, since_id));
					}
					
					
					/**
					List<TrafficUpdate> alltweets_update_list = new ArrayList<TrafficUpdate>();
					since_rs   = since_stmt.executeQuery(
							String.format("select MAX(source_id) FROM all_tweets WHERE city_id=%d", (int)city_id) 
							);
					
					since_rs.first();
					res = since_rs.getString("MAX(source_id)");
					//if (res.compareTo("NULL")==0) {
					if (res == null) {
						TLog.log(Level.FINER, "(Fetching all of today's tweets)");
						
						alltweets_update_list.addAll(tweet_comm.getAllUpdatesOnDay(a_city, getDateInTwitterFormat(new Date(System.currentTimeMillis()))));
						TrafficUpdate metaupdate = new TrafficUpdate(this, System.currentTimeMillis(), "TYPE_META");
						metaupdate.addData("update_list", alltweets_update_list);
						tweet_comm.addUpdateToQueue(metaupdate);
						System.err.println(alltweets_update_list.size() + " new tweets for " + a_city.cityName);
					} else {
						int since_id = Integer.parseInt(res);
						TLog.log(Level.FINER, String.format("(Fetching all tweets only since id %d)", since_id));
					
						
						alltweets_update_list.addAll(tweet_comm.getAllUpdatesSinceId(a_city, since_id));
						TrafficUpdate metaupdate = new TrafficUpdate(this, System.currentTimeMillis(), "TYPE_META");
						metaupdate.addData("update_list", alltweets_update_list);
						tweet_comm.addUpdateToQueue(metaupdate);
						System.err.println(alltweets_update_list.size() + " new tweets for " + a_city.cityName);
					}
					**/
					
					since_rs.close();
					since_stmt.close();
					
				}
				
				updateTweetCountsForCity(a_city);
				
				TLog.log(Level.FINE, String.format("Completed update for city '%s'", city_name));
				
				lastCheckedCityId = (int) city_id;
				update_stmt = dbConn.createStatement();
				int update_result = update_stmt.executeUpdate("UPDATE cities SET updated_at=NOW() WHERE id=" + city_id);
				//System.err.println("lastCheckedCityId is now " + lastCheckedCityId);
				
				update_stmt.close();
				
				has_next = next_n_rs.next();
			}
			
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			TLog.log(Level.WARNING, e);
			throw new TrafficSourceException("DiyFluSource: MySQL error occurred during check");
		} finally {
			try {
				if (next_n_rs != null) next_n_rs.close();
				if (next_n_stmt != null) next_n_stmt.close();
				if (since_rs != null) since_rs.close();
				if (since_stmt != null) since_stmt.close();
				if (update_stmt != null) update_stmt.close();
				
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				TLog.log(Level.WARNING, e);
			}
			
		}
		
		/** add this back in when threads are working 
		catch (TrafficSourceException te) {
			throw te;
		} finally {
			// is this safe here?!?
			this.UPDATE_IN_PROGRESS = false;
		}
		**/
		
		System.gc();
		
		return update_list;
	}

	private void updateTweetCountsForCity(City aCity) throws TrafficSourceException {
		TLog.log(Level.FINE, "Determing total number of tweets for city " + aCity.cityName);
		
		TweetComm tweet_comm = new TweetComm(this);
		
		Date now_date   = new Date(System.currentTimeMillis());
		String mysql_date_str   = getDateInFormat(now_date, "yyyy-MM-dd HH:00:00");
		String twitter_date_str = getDateInTwitterFormat(now_date);
		
		//System.err.print("THE DATE HERE IS " + mysql_date_str);
		
		Statement tweetCountStmt=null;
		ResultSet tweetCountRs=null;
		try {
			tweetCountStmt = dbConn.createStatement();
			tweetCountRs   = tweetCountStmt.executeQuery(
					// the date format here is yyyy-mm-dd because this MySQL table is only tracking the date (not time)
					String.format("SELECT * FROM hourly_counts WHERE city_id=%s ORDER BY date", 
							String.valueOf(aCity.cityId))
					);
			
			tweetCountRs.last();
			
			List<Status> status_list = null;
			int last_found_id = -1;
			if (tweetCountRs.getRow() <= 0) {
				// no cached last_found_id, just get all from the current date...
				TLog.log(Level.FINEST, String.format("No records yet (at all!) for city %s Querying on today (%s).", aCity.cityName, twitter_date_str));
				status_list = tweet_comm.getAllStatusesOnDay(aCity, twitter_date_str);
			} else {
				//tweetCountRs.first();
				// cached last_found_id, only get tweets since that id
				last_found_id = tweetCountRs.getInt("last_found_id");
				if (last_found_id >= 0) {
					TLog.log(Level.FINEST, String.format("Found a record for city %d (date %s), querying since id %s", aCity.cityId, tweetCountRs.getString("date"), last_found_id ));
					status_list = tweet_comm.getAllStatusesSinceId(aCity, last_found_id);
				} else {
					TLog.log(Level.FINEST, String.format("Found record for city %d (date %s), but 0 tweets. Querying on todays date (%s).", aCity.cityId, tweetCountRs.getString("date"), twitter_date_str));
					status_list = tweet_comm.getAllStatusesOnDay(aCity, twitter_date_str);
				}
			}
		
			if (status_list.size()==0) {
				TrafficUpdate t = new TrafficUpdate(this, System.currentTimeMillis(), "TYPE_COUNT");
				t.addData("count", 0);
				
				t.addData("date", mysql_date_str);
				t.addData ("city_id", aCity.cityId);
				t.addData("last_found_id", last_found_id);
			
				TLog.log(Level.FINER, String.format("Found 0 tweets for city %d, last found is %d ", aCity.cityId, last_found_id));
				tweet_comm.addUpdateToQueue(t);
			}
			else {
			
				Iterator<Status> statusI = status_list.iterator();
				HashMap<String, ArrayList<Status>> tally = new HashMap<String, ArrayList<Status>>();
				while (statusI.hasNext()) {
					Status next_status = statusI.next();
					//the date of the status is actually in local time, but want to convert to GMT so you can get
					//the current day there.
					//String tweet_date_str = DiyFluSource.getDateInTwitterFormat(next_status.createdAt); //
					String tweet_date_str  = DiyFluSource.getDateInFormat(next_status.createdAt, "yyyy-MM-dd HH:00:00");
					
					if (tally.containsKey(tweet_date_str)) {
						tally.get(tweet_date_str).add(next_status);
					} else {
						ArrayList<Status> new_list = new ArrayList<Status>();
						new_list.add(next_status);
						tally.put(tweet_date_str, new_list);
					}
				}
				
				Iterator<String> tally_iterator = tally.keySet().iterator();
				
				while (tally_iterator.hasNext()) {
					String next_date = tally_iterator.next();
					TrafficUpdate t = new TrafficUpdate(this, System.currentTimeMillis(), "TYPE_COUNT");
					t.addData("count", (tally.get(next_date)).size());
					t.addData("date", next_date);
					t.addData ("city_id", aCity.cityId);
					t.addData("last_found_id", String.valueOf(tally.get(next_date).get(0).id));
				
					TLog.log(Level.FINER, String.format("Found %s additional tweets in city %s for time %s", tally.get(next_date).size(), aCity.cityName, next_date));
					tweet_comm.addUpdateToQueue(t);
				}
				
				//TODO zero out tally here...
				tally_iterator = null;
				tally.clear();
				tally = null;
			}
			
		} catch (SQLException e) {
			TLog.log(Level.WARNING, e);
			throw new TrafficSourceException("MySQL error occurred while updating tweet counts for city!");
		} finally {
			try {
				if (tweetCountRs != null) tweetCountRs.close();
				if (tweetCountStmt != null) tweetCountStmt.close();
			} catch (SQLException e) {
				TLog.log(Level.WARNING, e);
			}	
		}
		
		System.gc();
	}
	
	private String[] fetchLatLongForCity(String cityName) throws TrafficSourceException {
		String coord[] = new String[2];
		//String latitude, longitude;
		
		ResultSet lat_long_rs = null; 
		Statement coord_stmt  = null;
		try {
			coord_stmt = dbConn.createStatement();

			lat_long_rs = coord_stmt.executeQuery(
					String.format("SELECT latitude, longitude FROM locations WHERE name=\"%s\"", cityName) 
					);
			
			
	
			lat_long_rs.last();
			if (lat_long_rs.getRow()<=0) {
				throw new TrafficSourceException("Can't find lat/long for city '" + cityName + "'" );
			}

			lat_long_rs.first();
			coord[0] = lat_long_rs.getString("latitude");
			coord[1] = lat_long_rs.getString("longitude");
			
			return coord;
			
		} catch (SQLException e) {
			TLog.log(Level.WARNING, e);
			throw new TrafficSourceException("SQL Error while fetching lat / longitude for '" +cityName+ "'");
		} finally {
			try {
				if (lat_long_rs != null) lat_long_rs.close();
				if (coord_stmt != null) coord_stmt.close();
			} catch (SQLException e) {
				TLog.log(Level.WARNING, e);
			}
			
		}
	
		
	}
	
	private Phrase[] fetchAllPhrases() throws TrafficSourceException {
		Phrase[] phrase_array = null;
		
		ResultSet phrase_rs   = null;
		Statement phrase_stmt = null;
		
		ResultSet query_rs   = null;
		Statement query_stmt = null;
		
		try {
			
			phrase_stmt = dbConn.createStatement();
			phrase_rs = phrase_stmt.executeQuery("Select id,title FROM phrases WHERE active=true");	
			
			phrase_rs.last();
			
			int num_rows = phrase_rs.getRow();
			TLog.log(Level.FINE, String.format("%d phrases are active", num_rows));
			
			phrase_array = new Phrase[num_rows];
			
			phrase_rs.beforeFirst();
			
			while(phrase_rs.next()) {
				query_stmt = dbConn.createStatement();
				query_rs = query_stmt.executeQuery(
						String.format("SELECT * from queries WHERE phrase_id=%d AND active=true", phrase_rs.getInt("id") )
						);
				
				query_rs.last();
				
				TLog.log(Level.FINE, String.format("%d queries are active for phrase %s", query_rs.getRow(), phrase_rs.getString("title")));
				
				Query[] query_array  = new Query[query_rs.getRow()];
				query_rs.beforeFirst();
				while (query_rs.next()) {
					query_array[query_rs.getRow()-1] = new Query(query_rs.getString("query"), query_rs.getInt("id"));
				}
				
				String phrase_title = phrase_rs.getString("title");
				long phrase_id      = phrase_rs.getLong("id");
			
				phrase_array[phrase_rs.getRow()-1] = new Phrase(phrase_title, phrase_id, query_array);
				
				query_rs.close();
				query_stmt.close();
				
			}
			
			phrase_rs.close();
			phrase_stmt.close();
			
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			TLog.log(Level.WARNING, e);
			throw new TrafficSourceException("MySQL error occurred while fetching phrases");
		} finally {
			try {
				if (phrase_stmt!=null) phrase_stmt.close();
				if (phrase_rs!=null) phrase_rs.close();
				if (query_stmt!=null) query_stmt.close();
				if (query_rs!=null) query_rs.close();
			} catch (SQLException sqle) {}
		}
		
		return phrase_array;
	}
	
	
	public String getResultForParams(HashMap<String,String> params, int page) throws FeedSourceException {
		
		InputStream istr = this.getFeedAsStream(params);
		
		// or InputStream sis  = new FileInputStream("ABC.txt");
		byte[] buffer = new byte[200000];
		ByteArrayOutputStream baos = new ByteArrayOutputStream(200000);
		int bytesread = 0;
		            
		while(true)
		{
		        try {
					bytesread = istr.read(buffer);
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		        if (bytesread == -1)
		             break;
		        baos.write(buffer, 0, bytesread);
		}
		
		byte[] stream_contents = baos.toByteArray();
		String result = new String(stream_contents);
		return result;
	}
	
	
	private static String getDateInTwitterFormat(Date dt) {
		return getDateInFormat(dt, "yyyy-MM-dd");
	}
	
	// does this  need to use 24 hours?!?
	private static String getDateInMySQLFormat(Date dt) {
		return getDateInFormat(dt, "yyyy-MM-dd HH:mm:ss");
	}
	
	private static String getDateInFormat(Date dt, String format) {
		
		SimpleDateFormat formatter = new SimpleDateFormat(format, 
				Locale.getDefault());
		
		// DO I WANT TO HAVE THIS IN HERE?!?
		formatter.setTimeZone(TimeZone.getTimeZone("GMT"));
		
		return formatter.format(dt);
	}
	
	/**
	 * Encodes a String by substituting a "+" in place of " "
	 * @param val a string to be plus-encoded
	 * @return a string which represents val, with all " " characters replaced by "+" characters
	 */
	private String plusEncode(String val){
		return val.replace(" ", "+");
	}
	
	private class City {
		public String cityName;
		public String countryName;
		public long   cityId;
		public String longitude;
		public String latitude;
		
		public City(String name, String country, long id, String lat, String lng) {
			this.cityName = name;
			this.countryName = country;
			this.cityId = id;
			this.latitude = lat;
			this.longitude = lng;
		}
	}
	
	private class Phrase {
		public String text;
		public long phraseId;
		private Query[] queries;
		
		public Phrase(String txt, long id, Query[] qs ) {
			this.text = txt;
			this.phraseId = id;
			this.queries = qs;
		}
	}
	
	private class Query {
		public String text;
		public long queryId;
		
		public Query(String txt, long id) {
			this.text = txt;
			this.queryId = id;
		}
	}
	
	private class Bounds {
		
		public final static String BOUND_TYPE_DATE = "DATE";
		public final static String BOUND_TYPE_ID = "ID";
		
		public HashMap<String,String> boundsParams;
		private String boundType;
		
		
		
		public Bounds() {
			this.boundsParams = new HashMap<String, String>();
		}
		
		public void addDateBounds(String since_date_str, String until_date_str) {
			this.boundType = Bounds.BOUND_TYPE_DATE;
			this.boundsParams.put("since", since_date_str);
			this.boundsParams.put("until", until_date_str);
		}
		
		public void addIdBounds(int id) {
			this.boundType = Bounds.BOUND_TYPE_ID;
			this.boundsParams.put("since_id", String.valueOf(id));
		}
	}
	
	// need to break out GoGetter into its own class.... with a generic method called "getWithParameters"
	// which takes a HashMap of params. Then create a getAllForCity(since id). Then access this from
	// another Source to get all tweets for a city. 
	// TODO But how to return both the number of tweets as well as id of the most recent tweet? (!!)
	// 		ANSWER: if you do return a list of TrafficUpdates, you can get both these pieces of information...
	// TODO Generic flag for intermediateUpdate (vs. returning a list)
	public class TweetComm implements Runnable {

		private City myCity;
		private TrafficSource mySource;
		private Phrase[] myPhrases;
		
		public TweetComm(TrafficSource source, City aCity, Phrase[] phrases) {
			mySource  = source;
			myCity    = aCity;
			myPhrases = phrases;
		}
		
		public TweetComm(TrafficSource source) {
			this(source, null, null);
		}
		
		/**
		public void getPhrasesSinceToday(City, Phrases[])
		
		public void getPhrasesSinceId(City, Phrases, id)
		public void getAllSinceId(City, id)
		
		public void getPhrasesSinceDaysAgo(City, Phrases, num_days)
		
		private getWithParams(params, drip)
		
		**/
		public List<TrafficUpdate> getPhrasesSinceDaysAgo(Phrase[] phrases, City aCity, int previousDays, boolean drip) {
			List<TrafficUpdate> update_list = new ArrayList<TrafficUpdate>();
						
			Calendar calendar_now = GregorianCalendar.getInstance(TimeZone.getDefault());
			calendar_now.setTimeInMillis(System.currentTimeMillis());
			
			TLog.log(Level.FINEST,String.format("Querying for phrases in city '%s' for %s days back", aCity.cityName, previousDays));
			
			/** goes back a certain number of days... **/
			for (int day_counter=0; day_counter<previousDays; day_counter++) {
				
				String use_date = DiyFluSource.getDateInTwitterFormat(calendar_now.getTime());
				List<TrafficUpdate> updates = getPhrasesOnDay(phrases, use_date, aCity);
				
				if (drip) {
					// add these to the queue
					addUpdatesToQueue(updates);
				} else {
					// do NOT add them to the update_list, don't want to add them twice!!
					update_list.addAll(updates);
				}
				
				calendar_now.add(Calendar.DATE, -1);
			} 
		
			return update_list;
		}
		
		public List<TrafficUpdate> getPhrasesSinceToday(Phrase[] phrases, City aCity) {
			Calendar calendar_now = GregorianCalendar.getInstance(TimeZone.getDefault());
			calendar_now.setTimeInMillis(System.currentTimeMillis());
			String use_date = DiyFluSource.getDateInTwitterFormat(calendar_now.getTime());
			
			return getPhrasesOnDay(phrases, use_date, aCity);
		}
		
		public List<TrafficUpdate> getPhrasesOnDay(Phrase[] phrases, String dateStr, City aCity) {
						
			HashMap<String, String> base_param_map = new HashMap<String, String>();
			base_param_map.put("since", dateStr);
			base_param_map.put("until", dateStr);
			base_param_map.put("geocode", String.format("%s,%s,15km", aCity.latitude, aCity.longitude));
			
			TLog.log(Level.FINEST,String.format("Querying for all phrases in city '%s' from date %s",  aCity.cityName, dateStr));
			return getPhrasesUsingBaseParams(phrases, base_param_map, aCity);
		}
		
		
		public List<TrafficUpdate> getPhrasesSinceId(Phrase[] phrases, City aCity, int since_id) {
			
			HashMap<String, String> base_param_map = new HashMap<String, String>();
			base_param_map.put("since_id", String.valueOf(since_id));
			base_param_map.put("geocode", String.format("%s,%s,15km", aCity.latitude, aCity.longitude));
			//param_map.put("q", phrase_query);
			
			TLog.log(Level.FINEST,String.format("Querying for terms all phrases in city '%s' since id %s", aCity.cityName, since_id));
			return getPhrasesUsingBaseParams(phrases, base_param_map, aCity );
		}

		
		public List<TrafficUpdate> getPhrasesUsingBaseParams(Phrase[] phrases, HashMap<String,String> base_params, City aCity) { 
			List<TrafficUpdate> updates	= new ArrayList<TrafficUpdate>();
			
			// iterate through all of the registered phrases
			for (int phr_count=0; phr_count< phrases.length; phr_count++) {
				//String phrase_query = buildPhraseQuery(phrases);
				
				Phrase a_phrase = phrases[phr_count];
				
				TLog.log(Level.FINEST, String.format("Processing phrase '%s' (%d/%d)", a_phrase.text, phr_count+1, phrases.length ));
				
				Query[] query_list = a_phrase.queries;
				
				// iterate through all of the queries for this phrase
				for (int qry_count=0; qry_count<query_list.length; qry_count++) {
					Query a_query = query_list[qry_count];
					
					HashMap<String, String> param_map = (HashMap<String, String>) base_params.clone();
					
					HashMap<String,String> query_params = parseQuery(a_query.text);
					Iterator<String> param_iter = query_params.keySet().iterator();
					
					// go through all the parameters for this query
					while (param_iter.hasNext()) {
						String param_key = (String) param_iter.next();
						param_map.put(param_key, query_params.get(param_key));
					}
					
					TLog.log(Level.FINEST,String.format("Searching on query '%s' (%d/%d)", printParams(param_map), qry_count+1, query_list.length));
					List<Status> statuses = fetchWithParameters(param_map);
					updates.addAll(createUpdatesFromStatuses(statuses, aCity, a_phrase, a_query, "TYPE_MENTION"));
				}
							
				
			}
			return updates;
		}
		
		/**
		public List<TrafficUpdate> getAllUpdatesOnDay(City aCity, String dateStr) {
			List<Status> status_list = getAllStatusesOnDay(aCity, dateStr);
			return createUpdatesFromStatuses(status_list, aCity, null, false, "TYPE_ALL");
		}
		**/
		
		public List<Status> getAllStatusesOnDay(City aCity, String dateStr) {
			
			HashMap<String, String> param_map = new HashMap<String, String>();
			param_map.put("since", dateStr);
			param_map.put("until", dateStr);
			param_map.put("geocode", String.format("%s,%s,15km", aCity.latitude, aCity.longitude));
			
							
			TLog.log(Level.FINEST,String.format("Querying for all tweets in city '%s' from date %s", aCity.cityName, dateStr));
			List<Status> statuses = fetchWithParameters(param_map);
			
			/**List<TrafficUpdate> updates	= createUpdatesFromStatuses(statuses, aCity, null);
			return updates;**/
			
			return statuses;
		}
		
		/**
		public List<TrafficUpdate> getAllUpdatesSinceId(City aCity, int since_id) {
			List<Status> status_list = getAllStatusesSinceId(aCity, since_id);
			return createUpdatesFromStatuses(status_list, aCity, null, false, "TYPE_ALL");
		}
		**/
		
		public List<Status> getAllStatusesSinceId(City aCity, int since_id) {
			
			HashMap<String, String> param_map = new HashMap<String, String>();
			param_map.put("since_id", String.valueOf(since_id));
			param_map.put("geocode", String.format("%s,%s,15km", aCity.latitude, aCity.longitude));
			
			TLog.log(Level.FINEST,String.format("Querying for all tweets in city '%s' since id %s", aCity.cityName, since_id));
			List<Status> statuses = fetchWithParameters(param_map);
			
			/**
			List<TrafficUpdate> updates	= createUpdatesFromStatuses(statuses, aCity, null);
			return updates;
			**/
			
			return statuses;
		}
		
				
		private List<TrafficUpdate> createUpdatesFromStatuses(List<Status> statuses, City aCity, Phrase phrs, Query qry, String updateType) {
			
			ArrayList<TrafficUpdate> traffic_updates = new ArrayList<TrafficUpdate>();
			
			Iterator<Status> statusI = statuses.iterator();
			
			while (statusI.hasNext()) {
				Status a_status = statusI.next();
				
				/**
				if (doFilter) {
					// filter out any statuses that have a link in them
					if (a_status.text.contains("http://")) continue;
					else if ((a_status.text.toLowerCase()).contains("swine flu")) continue;
				}
				**/
				
				TrafficUpdate t = new TrafficUpdate(this.mySource, System.currentTimeMillis(), updateType);
				
				t.addData("status_id", String.valueOf(a_status.id));
				
				t.addData("city_id", String.valueOf(aCity.cityId));
				t.addData("city_name", aCity.cityName);
				t.addData("country_name", aCity.countryName);
				
				t.addData("text", a_status.text);
				
				Date createdAt = a_status.createdAt;
				// TODO make sure this doesn't do a timeshift!!
				t.addData("mentioned_at", DiyFluSource.getDateInMySQLFormat(createdAt));
				
				String link_str = String.format("http://www.twitter.com/%s/statuses/%s",
						a_status.fromUser,
						a_status.id
				);
				// NEED TO GET THIS FROM TWITTER
				t.addData("link", link_str);
				t.addData("phrase_id", String.valueOf(phrs.phraseId));
				t.addData("query_id", String.valueOf(qry.queryId));
				
				/**
				if (phrases != null) {
					long phrase_id = -1;
					for (int phr_cnt=0; phr_cnt<phrases.length; phr_cnt++) {
						Phrase next_phrase = phrases[phr_cnt];
						if (a_status.text.toLowerCase().contains(next_phrase.text.toLowerCase())) {
							TLog.log(Level.FINEST, 
									String.format("Classifying tweet '%s' as phrase '%s' (%s)", a_status.text, next_phrase.text, next_phrase.phraseId)
									);
							phrase_id = next_phrase.phraseId;
							break;
						}
					}
					t.addData("phrase_id", String.valueOf(phrase_id));
				}
				**/
				
				t.addData("mentioner", a_status.fromUser);
			
				traffic_updates.add(t);
			}
			
			return traffic_updates;
		}
		
		private void addUpdatesToQueue(List<TrafficUpdate> updates) {
			Iterator<TrafficUpdate> i = updates.iterator();
			while (i.hasNext()) {
				addUpdateToQueue(i.next());
			}
		}
		
		private void addUpdateToQueue(TrafficUpdate tu) {
			try {
				AirTrafficController atc = AirTrafficController.getController();
				//if (atc!=null) {
					// i don't know why it would be null!
				atc.addUpdate(tu);
				//}
			} catch (IllegalStateException iae) {
				TLog.log(Level.WARNING, iae);
				TLog.log(Level.WARNING, "DiyFlu overflowed the update queue!");
			}
		}
		
		private String printParams(HashMap<String,String> params) {
			String q = "";
			Iterator i = params.keySet().iterator();
			
			while (i.hasNext()) {
				String key = (String) i.next();
				q += String.format("%s=%s", key,params.get(key));
				if (i.hasNext()) q+="&";
			}
			
			return q;
		}
		
		private HashMap<String,String> parseQuery(String q) {
			HashMap<String,String> query = new HashMap<String,String>();
			
			String[] params = q.split("&");
			for (int i=0; i< params.length; i++) {
				try {
					String[] key_val = params[i].split("=");
					query.put(key_val[0], key_val[1]);
				} catch (Exception e) {
					TLog.log(Level.WARNING, 
							String.format("Formatting error with query '%s'; skipping...", q)
							);
				}
			}
			
			return query;
		}
		
		private String buildPhraseQuery(Phrase[] phrases) {
			//need to concatenate the phrases into one search string, and build a map of phrases to phrase_id
			String phrase_search_str = "";
			//HashMap<String,String> phrase_id_map = new HashMap<String, String>();
			for (int phr_cnt=0; phr_cnt<phrases.length; phr_cnt++) {
				Phrase next_phrase = phrases[phr_cnt];
				//phrase_id_map.put(next_phrase.text, String.valueOf(next_phrase.phraseId));
				phrase_search_str += String.format("\"%s\"", plusEncode(next_phrase.text));
				if (phr_cnt < phrases.length-1) phrase_search_str += "+OR+";
			}
			
			return phrase_search_str;
		}

		/**
		private String describeQuery() {
			String bound_desc_str;
			if (params.containsKey("since")) {
				bound_desc_str = String.format(" date '%s' to date '%s'", params.get("since"), params.get("until"));
			} else {
				bound_desc_str = String.format(" id %s", params.get("since_id"));
			}
			//TLog.log(Level.FINER,String.format("Querying for terms '%s' in city '%s' from %s", params.get("q"), cty.cityName, bound_desc_str));
			return String.format("Querying for terms '%s' in city '%s' from %s", params.get("q"), cty.cityName, bound_desc_str));
		}
		**/
		
		/**
		 * Gets all the instances of a phrase in a city on a particular date
		 * @param cty A City wrapper object
		 * @param phr A Phrase wrapper object
		 * @param date_str A string representing the date on which to search
		 * @return
		 */
		
		
		/**
		 *  The atomic operation - one phrase for one city.
		 */
		public List<Status> fetchWithParameters(HashMap<String,String> params) {
			
			List<Status> all_status_list = new ArrayList<Status>();
				
			int page = 1;
			int attempts=1;
			final int MAX_ATTEMPTS = 20;
			
			// this loop will make 20 attempts
			while (true) {
				
				params.put("rpp", "100");
				
				params.put("page", String.valueOf(page));
				
				List<Status> status_list = null;
				try {
					//System.err.println("Attempt #" + attempts);
					String query_result = getResultForParams(params, page);
					try {
						status_list = Status.getStatuses(query_result);
					} catch (TwitterException e) { // if something happens parsing results
						TLog.log(Level.WARNING, e);
					}
				} catch (FeedSourceException e) { // if something happens while issuing request
					Level use_level = (attempts > 15) ? Level.WARNING : Level.FINE;
					
					//TLog.log(use_level, 
							//String.format(e.getMessage() + " ('%s' in %s from %s, attempt #%d)", phrase_search_str, cty.cityName, bound_desc_str, attempts)); 
					
					TLog.log(use_level, 
							String.format(e.getMessage() + " (attempt #%d)", attempts)); 
					
					// if you're getting 404s from Twitter, most likely you won't be able to go any
					// further back, so stop spamming them!
					if ( (attempts >= MAX_ATTEMPTS) || (e.statusCode == HttpStatus.SC_NOT_FOUND) ) {
						break;
					} else {
						attempts++;
						continue;
					}
				}

				// add the statuses to the list of /all/ statuses
				all_status_list.addAll(status_list);
				
				// get the number of statuses received this time around
				int num_statuses = status_list.size();
				
				// if it's less than the requested 100 ...
				// (should this just be if numNodes==0)?
				if (num_statuses < 100) {
					break;
				}
				
				page++;
			}
			
			return all_status_list;
		}
		
		
		@Override
		public void run() {
			getPhrasesSinceDaysAgo(this.myPhrases, this.myCity, 60, true);
		}
		
	}
}
